package personal.wang;

import org.apache.kafka.clients.admin.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @className: AdminClientTest
 * @Description:
 * @Author: wangyifei
 * @Date: 2025/4/22 9:53
 */
public class AdminClientTest {
    private static Logger logger = LoggerFactory.getLogger(AdminClientTest.class);

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka8:9092,kafka9:9092,kafka10:9092,kafka11:9092,kafka12:9092");
        props.put("request.timeout.ms", 600000);
        int[] par = {1,2,3,7,8,9,10,11,12};

        try (AdminClient client = AdminClient.create(props)) {
            ListTopicsResult listTopicsResult = client.listTopics();
            Set<String> topics = listTopicsResult.names().get();
            for (String topic : topics) {
                DescribeTopicsResult describeTopicsResult = client.describeTopics(Collections.singletonList(topic));
                Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
                stringTopicDescriptionMap.forEach((k,v)->{
                    if(!v.isInternal()){
                        String collect = IntStream.range(0, v.partitions().size())
                                .filter(x -> v.partitions().get(x).replicas().size() != v.partitions().get(x).isr().size())
                                .mapToObj(y -> String.format("{\"topic\":\"%s\",\"partition\":%s , \"replicas\":[%s]}"
                                        , k
                                        , v.partitions().get(y).partition()
                                        , v.partitions().get(y)
                                                .isr()
                                                .stream()
                                                .map(e -> e.idString())
                                                .collect(Collectors.joining(","))
                                                + "," + par[y % par.length]
                                )).collect(Collectors.joining(",\n"));
//                        String collect = v.partitions()
//                                .stream()
//                                .filter(x->x.replicas().size() != x.isr().size())
//                                .map(y -> String.format("{\"topic\":\"%s\",\"partition\":%s , \"replicas\":[%s]}"
//                                                          , k
//                                                          , y.partition()
//                                                          , y.isr().stream().map(e->e.idString()).collect(Collectors.joining(",\n") ) + "\n," + par[random.nextInt(par.length)]
//                                )).collect(Collectors.joining(","));
                            if(!collect.isEmpty()){
                                System.out.println(collect);
                            }
//                        });
                    }
                });

            }
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
